package com.xhdd.util;


import co.elastic.clients.elasticsearch.ElasticsearchClient;
import co.elastic.clients.elasticsearch._types.FieldValue;
import co.elastic.clients.elasticsearch._types.aggregations.Aggregate;
import co.elastic.clients.elasticsearch._types.aggregations.CardinalityAggregate;
import co.elastic.clients.elasticsearch._types.query_dsl.*;
import co.elastic.clients.elasticsearch.core.IndexResponse;
import co.elastic.clients.elasticsearch.core.ScrollResponse;
import co.elastic.clients.elasticsearch.core.SearchResponse;
import co.elastic.clients.elasticsearch.core.search.Hit;
import co.elastic.clients.elasticsearch.core.search.ResponseBody;
import co.elastic.clients.elasticsearch.indices.IndexState;
import co.elastic.clients.json.JsonData;
import co.elastic.clients.transport.endpoints.BooleanResponse;
import com.fasterxml.jackson.databind.JsonNode;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.fasterxml.jackson.databind.node.ArrayNode;
import com.xhdd.util.global.Params;
import org.springframework.stereotype.Component;

import javax.annotation.Resource;
import java.io.ByteArrayInputStream;
import java.io.IOException;
import java.io.StringReader;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

import static com.xhdd.constant.Const.*;

@Component
public class EsUtil {

    @Resource
    ElasticsearchClient esClient;
    @Resource
    ObjectMapper jMapper;

    public boolean indexDelete(String indexName) throws IOException {
        return esClient.indices().delete(d -> d.index(indexName)).acknowledged();
    }

    public boolean indexExists(String indexName) throws IOException {
        BooleanResponse exists = esClient.indices().exists(e -> e.index(indexName));
        return exists.value();
    }

    /*返回所有索引信息*/
    public Map<String, IndexState> listIndex() throws IOException {

        return esClient.indices().get(req -> req.index(ALL)).result();
    }


    public String insertDoc(String indexName, String id, JsonNode doc) throws IOException {
        ByteArrayInputStream inputStream = new ByteArrayInputStream(jMapper.writeValueAsBytes(doc));
        IndexResponse response = esClient.index(i -> {
            if (id != null) i.index(indexName).id(id).withJson(inputStream);
            i.index(indexName).withJson(inputStream);
            return i;
        });
        return response.id();
    }

    public boolean createIndex(String indexName, JsonNode mapping) throws IOException {
        StringReader stringReader = new StringReader(mapping.toPrettyString());
        return esClient.indices().create(c -> c.index(indexName).withJson(stringReader)).acknowledged();
    }


    public ArrayNode searchPrefix(String indexName, List<String> prefixs, Integer page, Integer size) throws IOException {
        Query query = getPrefixQuery(prefixs, null)._toQuery();
        SearchResponse<JsonNode> products = esClient.search(i -> i.index(indexName).query(query).from(page).size(size), JsonNode.class);
        return getResult(products);
    }


    public ArrayNode search(String indexName, List<String> prefixs, List<Params> params, Integer page, Integer size) throws IOException {
        List<Query> queries = getQuery(params);
        Query query = getPrefixQuery(prefixs, queries)._toQuery();
        SearchResponse<JsonNode> products = esClient.search(i -> i.index(indexName).query(query).from(page).size(size), JsonNode.class);
        return getResult(products);
    }

    private List<Query> getQuery(List<Params> params) {
        List<Query> queries = new ArrayList<>();
        List<Params> eq = new ArrayList<>();
        List<Params> rang = new ArrayList<>();
        for (Params param : params) {
            String flag = param.getFlag();
            if (flag == null) eq.add(param);
            else rang.add(param);
        }
        Query eqQ = MatchQuery.of(m -> {
            for (Params p : eq) {
                m.field(p.getKey()).query(FieldValue.of(p.getValue()));
            }
            return m;
        })._toQuery();
        Query rangQ = RangeQuery.of(m -> {
            for (Params p : rang) {
                String key = p.getKey();
                Object value = p.getValue();
                String flag = p.getFlag();
                switch (flag) {
                    case GT:
                        m.field(key).gt(JsonData.of(value));
                    case GTE:
                        m.field(key).gte(JsonData.of(value));
                    case LTE:
                        m.field(key).lte(JsonData.of(value));
                    case LT:
                        m.field(key).lt(JsonData.of(value));
                }
            }
            return m;
        })._toQuery();
        queries.set(0, eqQ);
        queries.set(1, rangQ);
        return queries;
    }


    private BoolQuery getPrefixQuery(List<String> prefixs, List<Query> queries) {
        BoolQuery boolQuery;
        if (queries != null && !queries.isEmpty()) {
            boolQuery = BoolQuery.of(b -> {
                for (String p : prefixs) {
                    b.should(s -> s.matchPhrasePrefix(
                            k -> k.field(BDCODE).query(p)
                    ));
                }
                b.must(queries.get(0)).must(queries.get(1));
                return b;
            });
        } else {
            boolQuery = BoolQuery.of(b -> {
                for (String p : prefixs) {
                    b.should(s -> s.matchPhrasePrefix(
                            k -> k.field(BDCODE).query(p)
                    ));
                }
                return b;
            });
        }
        return boolQuery;
    }

    public ArrayNode searchDoc(String indexName, List<Params> params, Integer page, Integer size) throws IOException {
        SearchResponse<JsonNode> products;
        if (!params.isEmpty()) {
            List<Query> query = getQuery(params);
            Query eqQ = query.get(0);
            Query rangQ = query.get(1);
            products = esClient.search(s -> s.index(indexName).query(q -> q.bool(b -> b.must(eqQ).must(rangQ))).from(page).size(size), JsonNode.class);
        } else {
            products = esClient.search(s -> s.index(indexName).from(page).size(size), JsonNode.class);
        }
        return getResult(products);
    }

    /*
     * from-size
     * es对深度翻页支持不好，当查询最后一页时相当于查询所有数据
     * 使用scroll游标查询，对查询结果进行缓存，使用scrollId进行翻页，scroll不支持跳页只能连续翻页
     */
    public Map searchRange(String indexName, String scrollId, Integer size) throws IOException {
        Map<String, Object> map = new HashMap<>();
        if (scrollId.isEmpty()) {
            SearchResponse<JsonNode> products = esClient.search(s -> s.index(indexName)
//                    .query()  搜索条件
                    .scroll(t -> t.time("5m"))
                    .size(size), JsonNode.class);
            String id = products.scrollId();
            ArrayNode result = getResult(products);
            map.put("id", id);
            map.put("result", result);

        } else {
            ScrollResponse<JsonNode> scroll = esClient.scroll(s -> s.scrollId(scrollId).scroll(t -> t.time("5m")), JsonNode.class);
            String id = scroll.scrollId();
            ArrayNode result = getResult(scroll);
            map.put("id", id);
            map.put("result", result);
        }
        return map;
    }

    public void clearScroll(ElasticsearchClient esClient) throws IOException {
        esClient.clearScroll();
    }

    public ArrayNode getAllDoc(String indexName) throws IOException {
        SearchResponse<JsonNode> products = esClient.search(s -> s.index(indexName), JsonNode.class);
        return getResult(products);
    }

    ArrayNode getResult(ResponseBody<JsonNode> products) {
        ArrayNode retn = jMapper.createArrayNode();
        for (Hit<JsonNode> hit : products.hits().hits()) {
            JsonNode jsonObject = hit.source();
            retn.add(jsonObject);
        }
        return retn;
    }


    /*求某个属性的最大值、最小值、平均值*/
    public Map<String, Double> propAggs(String indexName, String prop) throws IOException {
        SearchResponse<Map> search = esClient.search(s -> s.index(indexName)
                .aggregations(STATS, a -> a.stats(m -> m.field(prop))), Map.class);
        Map<String, Aggregate> aggregations = search.aggregations();
        Aggregate stats = aggregations.get(STATS);
        Map<String, Double> res = new HashMap<>();
        res.put(MAX, stats.stats().max());
        res.put(MIN, stats.stats().min());
        res.put(AVG, stats.stats().avg());
        return res;
    }

    /*去重统计在非分词类型上结果准确*/
    public long noRepeatCount(String indexName, String prop) throws IOException {
        SearchResponse<Map> search = esClient.search(s -> s.index(indexName)
                .aggregations(CARDINALITY, a -> a.cardinality(m -> m.field(prop))), Map.class);
        Map<String, Aggregate> aggregations = search.aggregations();
        Aggregate stats = aggregations.get(CARDINALITY);
        CardinalityAggregate cardinality = stats.cardinality();
        return cardinality.value();
    }


    /*分组，热力图*/
    public HashMap<String, Long> groupBy(String indexName, List<Params> params, String prop, String type) throws IOException {
        HashMap<String, Long> res = new HashMap<>();
        SearchResponse<Map> search;
        if (!params.isEmpty()) {
            List<Query> query = getQuery(params);
            Query eqQ = query.get(0);
            Query rangQ = query.get(1);
            search = esClient.search(s -> s.index(indexName).query(q -> q.bool(b -> b.must(eqQ).must(rangQ)))
                    .aggregations(GROUPBY, a -> a.terms(m -> m.field(prop))), Map.class);
        } else {
            search = esClient.search(s -> s.index(indexName)
                    .aggregations(GROUPBY, a -> a.terms(m -> m.field(prop))), Map.class);
        }

        Map<String, Aggregate> aggregations = search.aggregations();
        Aggregate aggregate = aggregations.get(GROUPBY);
        /*   .lterms()要注意类型*/
        if (LONG.equals(type)) {
            aggregate.lterms().buckets().array().forEach(bucket -> {
                res.put(bucket.keyAsString(), bucket.docCount());
            });
        }
        if (STRING.equals(type)) {
            aggregate.sterms().buckets().array().forEach(bucket -> {
                res.put(bucket.key().stringValue(), bucket.docCount());
            });
        }
        if (DOUBLE.equals(type)) {
            aggregate.dterms().buckets().array().forEach(bucket -> {
                res.put(bucket.keyAsString(), bucket.docCount());
            });
        }
        return res;
    }


}